﻿using System;
using System.Collections.Generic;
using System.Diagnostics;
using System.Linq;
using System.Reactive.Linq;
using System.Reactive.Subjects;
using System.Threading;
using System.Threading.Tasks;
using Apache.IoTDB;
using Apache.IoTDB.DataStructure;
using Iot.Shared.Helper;
using Iot.Shared.Path;
using Iot.TimingDatabase.Helper;
using Iot.TimingDatabase.Models;
using Mediator;
using Microsoft.Extensions.Logging;

namespace Iot.TimingDatabase.Event;

/// <summary>
///     设备模块数据记录到数据库
/// </summary>
public sealed class DeviceModuleDataToDbEvent : INotificationHandler<DeviceModule>
{
    private readonly IotDbConversion _iotDbConversion;

    /// <summary>
    ///     日志记录器
    /// </summary>
    private readonly ILogger<DeviceModuleDataToDbEvent> _logger;

    private readonly SessionPool _session;


    public readonly Subject<(string deviceId, RowRecord data)> BufferedSubject = new();

    public DeviceModuleDataToDbEvent(SessionPool session, IotDbConversion iotDbConversion,
        ILogger<DeviceModuleDataToDbEvent> logger)
    {
        _logger = logger;
        _session = session;
        _iotDbConversion = iotDbConversion;
        // 使用缓冲流，每5秒或当缓冲区达到1000个项目时触发
        // Where操作符过滤掉空集合
        // 订阅并处理数据归档操作
        BufferedSubject.Buffer(TimeSpan.FromSeconds(5), 1000) // 设置缓冲时间为5秒，缓冲区大小为1000
            .Where(static a => a.Count > 0) // 过滤条件：只保留数量大于0的集合
            .Subscribe(DataArchiving); // 订阅数据归档方法
    }

    /// <summary>
    ///     实现INotificationHandler接口的Handle方法，用于处理设备模块的通知
    /// </summary>
    /// <param name="notification"></param>
    /// <param name="cancellationToken"></param>
    /// <returns></returns>
    public ValueTask Handle(DeviceModule notification, CancellationToken cancellationToken)
    {
        try
        {
            _logger.LogTrace("记录设备模块数据到数据库，设备模块为：{notification}", notification);
            // 判断数据库连接是否打开
            if (_session.IsOpen())
            {
                var debugStopwatch = Stopwatch.StartNew(); // 创建并启动计时器
                // 获取设备模块的扩展值
                var data = notification.GetExtendedValues();
                _logger.LogTrace("记录设备模块数据到数据库，设备模块为：{notification},数据为：{data}", notification, data);
                // 将数据分发到数据库
                DistributeDataToDb(notification, data);
                debugStopwatch.Stop();
                _logger.LogTrace("记录设备模块数据到数据库，设备模块为：{notification},耗时：{time}", notification, debugStopwatch.Elapsed);
            }
            else
            {
                _logger.LogTrace("记录设备模块数据到数据库，设备模块为：{notification},数据库连接未打开", notification);
            }
        }
        catch (Exception e)
        {
            _logger.LogError(e, "设备模块数据记录到数据库，设备模块为：{notification}", notification);
        }

        return ValueTask.CompletedTask;
    }

    /// <summary>
    ///     数据记录到数据库
    /// </summary>
    /// <param name="data"></param>
    /// <param name="values"></param>
    /// <returns></returns>
    private void DistributeDataToDb(DeviceModule data,
        IReadOnlyList<ExtendedValue> values)
    {
        var deviceId = data.Path.ToString(); // 获取设备ID
        if (!deviceId.IsOriginalRootPath())
        {
            _logger.LogTrace("设备ID不以\"root.\"开头，不进行数据归档，，类型为：{data},值为：{values}", data, values);
            return;
        }

        _logger.LogTrace("数据记录到数据库，类型为：{data},值为：{values}", data, values);
        var iotData = values.Where(static a => a.Value is not null)
            .Select(a => (a.Name, Value: _iotDbConversion.DeviceModuleDataToDb(data.Path, a.Name, a.Value))).ToArray();
        if (iotData.Length <= 0) return;
        // 创建行记录
        var rowRecord = new RowRecord(data.TimeStamp.GetUnixTimeMilliseconds(),
            [.. iotData.Select(static a => a.Value.Value)],
            [.. iotData.Select(static a => a.Name)],
            iotData.Select(static a => a.Value.Type).ToList());
        // 调用数据归档方法
        _logger.LogTrace("开始调用数据归档方法，设备ID为：{deviceId},数据为：{data}", data.Path, rowRecord);
        DataArchiving(deviceId, rowRecord);
    }

    /// <summary>
    ///     数据归档
    /// </summary>
    /// <param name="deviceId"></param>
    /// <param name="data"></param>
    /// <returns></returns>
    private void DataArchiving(string deviceId, RowRecord data)
    {
        if (!deviceId.IsOriginalRootPath())
        {
            _logger.LogTrace("设备ID不以\"root.\"开头，不进行数据归档，设备ID为：{deviceId}", deviceId);
            return;
        }

        try
        {
            BufferedSubject.OnNext((deviceId, data));
            _logger.LogTrace("缓存数据成功，设备ID为：{deviceId},数据为：{data}", deviceId, data);
        }
        catch (Exception e)
        {
            _logger.LogError(e, "缓存数据失败，设备ID为：{deviceId},数据为：{data}", deviceId, data);
        }
    }

    private void DataArchiving(IList<(string deviceId, RowRecord data)> data)
    {
        var deviceIds = data.Select(static a => a.deviceId).ToList();
        var rowRecords = data.Select(static a => a.data).ToList();
        ThreadHelper.RunTask(async () =>
        {
            var code = await _session.InsertAlignedRecordsAsync(deviceIds, rowRecords);
            _logger.LogTrace("数据归档{result}，数据为：{data},结果为：{code}", code == 0 ? "成功" : "失败", data, code);
        }, logger: _logger);
    }
}